package test.zipParser;

import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileFilter;
import java.io.FileInputStream;
import java.io.IOException;
import java.net.URI;
import java.text.SimpleDateFormat;
import java.util.Calendar;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.util.GenericOptionsParser;

public class UploadZipFiles implements FileFilter{
	private String srcUri;
	private String targetUri;
	
	public UploadZipFiles(String srcUri,String targetUri){
		this.srcUri=srcUri;
		if(!targetUri.endsWith("/")&&!targetUri.endsWith("\\"))
			this.targetUri=targetUri+"/";
		else
			this.targetUri=targetUri;
	}
	public  void UploadData(String filename,FileFilter filter) throws IOException{
		try{
		Configuration conf=new Configuration();
		String target=targetUri+filename;
		FileSystem fs=FileSystem.get(URI.create(target), conf);
		System.out.println("target uri:"+target);
		BytesWritable value=new BytesWritable();
		Text key=new Text();
		SequenceFile.Writer writer=null;
		Path seqPath=new Path(target);
		writer=SequenceFile.createWriter(fs, conf, seqPath, Text.class, 
				BytesWritable.class);
		File file=new File(srcUri);
		File[] files=file.listFiles(filter);
		for(int i=0;i<files.length;i++){
			BufferedInputStream bis=null;
		 bis=new BufferedInputStream(new FileInputStream(files[i]));
			long len=files[i].length();
			byte[] buff=new byte[(int)len];
			IOUtils.readFully(bis, buff, 0, (int) len);
			//bis.read(buff, 0, (int)len);
			//value.readFields(new DataInputStream(bis));
			value.set(buff,0,(int)len);
			key.set(files[i].getName());
			
			writer.append(key, value);
			bis.close();
			System.out.println("path:"+files[i].getName()+" length:"+len);
			}
		IOUtils.closeStream(writer);
		}catch(Exception e){
			e.printStackTrace();
			System.exit(1);
		//	System.setOut(new PrintStream(new FileOutputStream("/home/hadoop/hadoopLogs.txt")));
		}
		}
	@Override
	public boolean accept(File pathname) {
		return pathname.getName().endsWith(".zip");
	}
	
	
	public static void main(String[] args) {
	if (args.length != 2 ){
		System.err.println("Usage:  <srcUri> <tmpLocalPath>");
		System.exit(2);
	}
		UploadZipFiles upZip = new UploadZipFiles(args[0],
				args[1]);
		SimpleDateFormat sf = new SimpleDateFormat("yyyyMMddHHmmssSSS");
		String newFileName = sf.format(Calendar.getInstance().getTime());
		try {
			upZip.UploadData(newFileName + ".seq", upZip);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}
